package com.boot.mapreduce;

import java.text.Collator;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

public class Coordinator {
    private Job currentJob;
    private List<Worker> workers = new ArrayList<>();
    private Queue<Job.Task> taskQueue = new LinkedList<>();
    private ExecutorService workerPool;

    public void submitJob(Job job) {
        this.currentJob = job;
        //创建Map任务，默认3个
        for (int i = 0; i < 3; i++) {
            Job.Task task = new Job.Task();
            task.type  = Job.Task.Type.MAP;
            task.taskId = i;
            taskQueue.add(task);
        }

        //启动Worker，默认2个
        startWorkers(2);
        //等待Map任务全部完成
        while(taskQueue.stream().anyMatch(t -> t.type == Job.Task.Type.MAP)) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.out.println("中断当前进程");
            }
        }


        //创建Reduce任务，默认2个
        for (int i = 0; i < 2; i++) {
            Job.Task task = new Job.Task();
            task.type  = Job.Task.Type.REDUCE;
            task.taskId = i;
            taskQueue.add(task);
        }

        //等待Reduce任务全部完成
        while(taskQueue.stream().anyMatch(t -> t.type == Job.Task.Type.REDUCE)) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.out.println("中断当前进程");
            }
        }

        workerPool.shutdown();

    }

    public Job getJob() {
        return currentJob;
    }

    public void startWorkers(int numWorkers) {
        //创建进程池
        workerPool = Executors.newFixedThreadPool(numWorkers);
        for (int i = 0; i < numWorkers; i++) {
            Worker worker = new Worker(this, i);
            workers.add(worker);
            workerPool.execute(worker);
        }
    }

    // Worker调用：获取任务
    public synchronized Job.Task getNextTask() {
        return taskQueue.poll();
    }

    // Worker调用：处理中间数据（简化版直接内存存储）
    public synchronized void storeIntermediateResult(int taskId, List<Pair<String, Integer>> results) {
        Map<String, List<Integer>> collect = results.stream().collect(Collectors.groupingBy(Pair::getKey, Collectors.mapping(Pair::getValue, Collectors.toList())));
        List list = collect.entrySet().stream().map(entry -> new Pair(entry.getKey(), entry.getValue())).collect(Collectors.toList());
        System.out.println(list);
        System.out.printf("Worker stored %d intermediate results from task %d\n", results.size(), taskId);
    }

    // 合并最终结果（简化版直接打印）
    public synchronized void mergeFinalResult(int taskId, List<Pair<String, Integer>> results) {
        System.out.println("=== Final Output ===");
        results.forEach(pair -> {
            System.out.println(pair.key + ": " + pair.value);
        });
    }
}
